package com.cl.spark.node;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cl.spark.base.BaseSparkNode;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.spark.enums.SparkNodeEnum;
import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.expressions.Window;
import org.apache.spark.sql.expressions.WindowSpec;
import org.apache.spark.sql.functions;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

// 去重取最新的
@Component
public class DropDuplicatesNode extends BaseSparkNode {

    @Override
    public SparkResult process(SparkParam sparkParam) {
        Dataset<Row> dataset = sparkParam.getSparkResultList().get(0).getDataset();
        JSONObject expression = sparkParam.getNodeExpression();

        JSONArray fields = expression.getJSONArray("fields");

        List<String> deduplicateFieldsList = new ArrayList<>();
        for (int i = 0; i < fields.size(); i++) {
            deduplicateFieldsList.add(fields.getString(i));
        }

        dataset = dataset.dropDuplicates(deduplicateFieldsList.toArray(new String[0]));
        return SparkResult.success(dataset);
    }


    @Override
    public SparkNodeEnum getType() {
        return SparkNodeEnum.DROP_DUPLICATES;
    }

}
